package beautiful.butterfly.distributed_task_processing_engine.zookeeper;

import beautiful.butterfly.distributed_task_processing_engine.tools.Constants;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

@Slf4j
public class ZooKeeperManager
{

    @Setter
    @Getter
    private ZooKeeper zooKeeper;
    @Setter
    @Getter
    private List<ACL> aclList = new ArrayList<ACL>();
    @Setter
    @Getter
    private Properties properties;
    @Setter
    @Getter
    private boolean isCheckParentLayHasScheduleTag = true;

    public ZooKeeperManager(Properties properties) throws Exception
    {
        this.properties = properties;
        try
        {
            this.connect();
        } catch (Exception e)
        {
            e.printStackTrace();
            throw e;
        }

    }


    public synchronized void reConnect() throws Exception
    {
        if (this.zooKeeper != null)
        {
            this.zooKeeper.close();
            this.zooKeeper = null;
            this.connect();
        }
    }

    private void connect() throws Exception
    {

        log.info("connect zookeeper active");
        connectZookeeper();
        log.info("connect zookeeper end");
        //
    }

    private void connectZookeeper() throws Exception
    {
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        zooKeeper = new ZooKeeper(this.properties.getProperty(keys.zooKeeperConnectString.toString()), //
                Integer.parseInt(this.properties.getProperty(keys.zooKeeperSessionTimeout.toString())),//
                new Watcher()
                {//
                    public void process(WatchedEvent watchedEvent)
                    {
                        sessionEvent(countDownLatch, watchedEvent);
                    }
                });
        //只关心连接成功
        countDownLatch.await();
        addAuthInfo();

        //aclList.add(new ACL(ZooDefs.Perms.READ, Ids.ANYONE_ID_UNSAFE));

    }

    private void addAuthInfo() throws Exception
    {
        //
        String authString = this.properties.getProperty(keys.userName.toString()) + ":" + this.properties.getProperty(keys.password.toString());
        this.isCheckParentLayHasScheduleTag = Boolean.parseBoolean(this.properties.getProperty(keys.isCheckParentPathVersionIsCompatible.toString(), "true"));
        zooKeeper.addAuthInfo("digest", authString.getBytes());
        aclList.clear();
        aclList.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(authString))));
    }

    private void sessionEvent(CountDownLatch countDownLatch, WatchedEvent watchedEvent)
    {
        log.warn("sessionEvent active");
        log.warn("watchedEvent.getState();" + watchedEvent.getState());
        if (watchedEvent.getState() == KeeperState.SyncConnected)
        {
            countDownLatch.countDown();
        } else if (watchedEvent.getState() == KeeperState.Expired)
        {
            try
            {
                reConnect();
            } catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.Disconnected)
        {
            try
            {
                reConnect();
            } catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.NoSyncConnected)
        {
            try

            {
                reConnect();
            } catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        } else
        {
            log.warn("event.getState() =" + watchedEvent.getState() + ", event  value=" + watchedEvent.toString());
            countDownLatch.countDown();
            throw new IllegalArgumentException("非法状态");
        }
        log.warn("sessionEvent end");
    }

    public void close() throws InterruptedException
    {
        log.info("close zookeeper");
        if (zooKeeper == null)
        {
            return;
        }
        this.zooKeeper.close();
    }

    public String getRootPath()
    {
        return this.properties.getProperty(keys.rootPath.toString());
    }

    public String getZooKeeperConnectString()
    {
        return this.properties.getProperty(keys.zooKeeperConnectString.toString());
    }

    public boolean zooKeeperIsConnected()
    {
        if (!Constants.debug)
        {
            log.info("zooKeeper.getState() :" + zooKeeper.getState());
        }
        return zooKeeper != null && zooKeeper.getState() == States.CONNECTED;
    }

    public void checkParentPathHasScheduleTagAndCheckTaskEngineVersion() throws Exception
    {
        log.info("创建root节点,兼容性检测");
        if (zooKeeper.exists(this.getRootPath(), false) == null)
        {
            ZooKeepers.createPath(zooKeeper, this.getRootPath(), aclList, CreateMode.PERSISTENT);
            if (isCheckParentLayHasScheduleTag == true)
            {
                checkParentPathHasScheduleTag(zooKeeper, this.getRootPath());
            }
            zooKeeper.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
        } else
        {
            if (isCheckParentLayHasScheduleTag == true)
            {
                checkParentPathHasScheduleTag(zooKeeper, this.getRootPath());
            }
            byte[] bytes = zooKeeper.getData(this.getRootPath(), false, null);
            if (bytes == null)
            {
                zooKeeper.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
            } else
            {
                String version = new String(bytes);
                log.info("" + Version.getVersion() + " " + version);
                if (Version.isCompatible(version) == false)
                {
                    throw new Exception("版本不一致:" + Version.getVersion() + " Zookeeper " + version);
                }

            }
        }
    }

    /**
     * 检查父级不能带有schedule-
     */
    public static void checkParentPathHasScheduleTag(ZooKeeper zooKeeper, String path) throws Exception
    {
        String[] strings = path.split(Constants.slash);
        String temporaryPath = "";
        for (int i = 0; i < strings.length - 1; i++)
        {
            String string = strings[i];
            if (!string.equals(""))
            {
                temporaryPath = temporaryPath + Constants.slash + string;
                if (zooKeeper.exists(temporaryPath, false) != null)
                {
                    byte[] bytes = zooKeeper.getData(temporaryPath, false, null);
                    if (bytes != null)
                    {
                        String value = new String(bytes);
                        if (value.indexOf("distributed_task_processing_engine-") >= 0)
                        {
                            throw new Exception("\"" + temporaryPath + "\"  is already a distributed_task_processing_engine instance's root directory, its any subdirectory cannot as the root directory of others");
                        }
                    }
                }
            }
        }
    }

    //


    public ZooKeeper getZooKeeper() throws Exception
    {
        if (this.zooKeeperIsConnected() == false)
        {
            reConnect();
        }
        return this.zooKeeper;
    }

}
